package lucene.server.cdc.client;

import lucene.server.common.storage.StorageHelper;
import lucene.server.common.utils.SecretUtil;
import lucene.server.common.utils.StringUtil;
import lucene.server.database.DBMetaService;
import lucene.server.database.model.ColumnModel;
import lucene.server.database.model.DatabaseModel;
import lucene.server.database.model.TableModel;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class Filter {
    private Properties properties;
    private List<String> formatedFilter;
    private List<String> databaseFilter = new ArrayList<>();
    private ConcurrentHashMap<String, List<ColumnModel>> tableFilter ;


    public Filter(Properties properties){
        this.properties = properties;
        this.formatedFilter =  getFormatedFilter(properties.getProperty("filter"));
        this.databaseFilter =  new ArrayList<>();
        String metaMapKey =  SecretUtil.generateMD5(properties.getProperty("filter"));
        String filterMapPath = properties.get("name")+metaMapKey+"_filter.cdc";

        this.tableFilter = (ConcurrentHashMap<String, List<ColumnModel>>) StorageHelper.fromFile(filterMapPath);
        if (this.tableFilter == null){
            this.tableFilter = new ConcurrentHashMap<>();
            createFilter();
            StorageHelper.toFile(this.tableFilter,filterMapPath);
        }
    }
    public List<ColumnModel> getTableColumns(String key){
        return this.tableFilter.get(key);
    }
    public boolean containsTable(String databaseName,String tableName) {
        return this.tableFilter.containsKey(databaseName+"::"+tableName);
    }
    public boolean containsDatabase(String databaseName) {
       return  this.databaseFilter.contains(databaseName);
    }
    //todo
    public void refreshColumn(String databaseName,String tableName) {
        String connName = "cdc_"+this.properties.getProperty("name");
        String dbType   = this.properties.getProperty("type");
        List<ColumnModel> columnModels =  DBMetaService.getColumns(connName,dbType, databaseName, tableName  );
        this.tableFilter.remove(databaseName+"::"+tableName);
        this.tableFilter.put(databaseName+"::"+tableName, columnModels);
     }
    public  void createFilter() {

        String connName = "cdc_"+this.properties.getProperty("name");
        String dbType   = this.properties.getProperty("type");
        // 获取所有表，格式 数据库::表名
        Map<String, DatabaseModel> databaseModelMap = DBMetaService.getDatabases(connName,dbType);
        createDatabaseFilter(connName,dbType,databaseModelMap);
        createTableFilter(connName,dbType);

    }
    /**初始化数据库过滤器*/
    private void createDatabaseFilter(String connName,String connType,Map<String, DatabaseModel> databaseModelMap) {

        if(this.formatedFilter.contains("*.*")){
            databaseModelMap.keySet().forEach(dbName -> {
                this.databaseFilter.add(dbName);
            });
        }else{
            this.formatedFilter.forEach(filter -> {
                String dbName = filter.split("\\.")[0];
                if (!this.databaseFilter.contains(dbName)){
                    this.databaseFilter.add(dbName);
                }
            });
        }
    }
    /**初始化表过滤器*/
    private void createTableFilter(String connName,String connType) {

        for (String database : this.databaseFilter){
            Map<String,  TableModel>  allTables = DBMetaService.getTables(connName,connType,database);
            for (String tableName : allTables.keySet())
            {
                for (String filter : this.formatedFilter) {
                    String[] filterItems = filter.split("\\.");
                    String itemDb = filter.split("\\.")[0];
                    String itemTable = filterItems.length > 1 ? filterItems[1] : "*";
                    if (database.equals(itemDb) && (itemTable.equals("*") || tableName.equals(itemTable))) {
                        List<ColumnModel> columnModels = DBMetaService.getColumns(connName,connType, database, tableName);
                        this.tableFilter.put(database + "::" + tableName, columnModels);
                    }
                }
            }
        }
    }

    public static List<String> getFormatedFilter(String filterStr) {

        List<String> filterList = new ArrayList<>();
        if (StringUtil.isNotEmpty(filterStr)) {
            filterList = new ArrayList<>(Arrays.asList(filterStr.split(",")));
            filterList.removeIf(String::isEmpty);
            // 遍历列表并修改
            for (int i = 0; i < filterList.size(); i++) {
                String filter = filterList.get(i);
                if (!filter.contains(".")) {
                    filterList.set(i, filter + ".*");
                } else if (filter.endsWith(".")) {
                    filterList.set(i, filter + "*");
                }
            }
        } else {
            filterList.add("*.*");
        }
        return filterList;
    }
}
